package com.sweet.logging.storage.store;

import com.sweet.logging.storage.StorageService;
import com.sweet.logging.storage.entity.LogMessage;

import javax.naming.ConfigurationException;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.Charset;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

/**
 * 文件存储
 *    将日志数据存储在文件中，同时创建时间戳查询索引
 *
 * @author xiexiaofeng
 * @version 1.0     2017年09月09日  18点38分
 */
public class SimpleFileStorageService implements StorageService
{
    public static final String INDEX_NAME      = "sweet.index";      // 时间戳索引文件名
    public static final String INDEX_HASH_NAME = "sweet_hash.index"; // hash索引文件名
    public static final String DATA_FILE_NAME  = "sweet.log";        // 日志文件名

    private Charset charset = Charset.forName("UTF-8");

    private File   indexFile     = null;
    private File   hashIndexFile = null;
    private File   dataFile      = null;
    private String rootPath      = null;

    private FileChannel indexChannel     = null;
    private FileChannel hashIndexChannel = null;
    private FileChannel dataChannel      = null;

    private FileOutputStream indexOutputStream     = null;
    private FileOutputStream hashIndexOutputStream = null;
    private FileOutputStream dataOutputStream      = null;

    private TreeMap<Long, Long> indexTree = new TreeMap<>(); // 索引

    /**
     * 初始化，打开数据文件和索引文件
     * @param logRootDir 存储根目录
     * @throws ConfigurationException 配置异常，目录无效等问题
     * @throws IOException 文件打开失败，文件损坏，文件被锁住等问题
     */
    public SimpleFileStorageService(String logRootDir) throws ConfigurationException, IOException
    {
        File storageDir = new File(logRootDir);
        if (!storageDir.exists())
        {
            boolean success = storageDir.mkdirs();
            if (!success)
                throw new IOException("create directory (" + logRootDir + ") failed");
        }
        if (!storageDir.isDirectory())
        {
            throw new ConfigurationException("storageDir not set or it`s wrong");
        }
        rootPath      = storageDir.toPath().toAbsolutePath().toString() + File.separator;
        dataFile      = new File(rootPath + DATA_FILE_NAME);
        indexFile     = new File(rootPath + INDEX_NAME);
        hashIndexFile = new File(rootPath + INDEX_HASH_NAME);
        if (!dataFile.exists())
        {
            if (!dataFile.createNewFile())
                throw new IOException("cant`t create log file " + dataFile.getAbsolutePath());
        }
        if (!indexFile.exists())
        {
            if (!indexFile.createNewFile())
                throw new IOException("cant`t create index file " + indexFile.getAbsolutePath());
        }
        if (!hashIndexFile.exists())
        {
            if (!hashIndexFile.createNewFile())
                throw new IOException("cant`t create index file " + hashIndexFile.getAbsolutePath());
        }
        dataOutputStream      = new FileOutputStream(dataFile, true);
        indexOutputStream     = new FileOutputStream(indexFile, true);
        hashIndexOutputStream = new FileOutputStream(hashIndexFile, true);
        dataChannel      = dataOutputStream.getChannel();
        indexChannel     = indexOutputStream.getChannel();
        hashIndexChannel = hashIndexOutputStream.getChannel();
    }

    private static final int BUFFER_SIZE = 2048; // 2K缓冲池，如果还不够用，就临时创建个更大的，避免浪费内存
    private ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE); // 缓冲

    /**
     * 批量保存日志数据，保证数据数据顺序写入，这里需要加锁
     * @param messages 日志数据
     */
    @Override
    public synchronized void save(List<LogMessage> messages)
    {
        for (LogMessage message : messages)
        {
            try
            {
                // 这里修复一个BUG，如果是不同机器，同一时间点发生的日志，应该保存下来的。这里需要通过时间戳和IP共同保证唯一性，防止丢数据。
                Long oldPosition = indexTree.putIfAbsent(message.getTimeStamp() & message.getClient().hashCode(), dataChannel.position());
                if (oldPosition == null) // 如果是新的时间点，需要保存索引的
                {
                    saveIndex(message.getTimeStamp(), dataChannel.position()); // 保存时间戳索引
                    if (indexTree.size() > 50000) // 防止它过大
                    {
                        TreeMap<Long, Long> treeMap = new TreeMap<>();
                        for (int i = 0; i < 10; i++)
                        {
                            // 这里的循环次数最等于接近集群中机器的数据量，这样才能保证同一时间戳的数据不会被覆盖
                            Map.Entry<Long, Long> entry = indexTree.pollLastEntry();
                            treeMap.put(entry.getKey(), entry.getValue());
                            indexTree = treeMap;
                            System.gc(); // 建议GC下
                        }
                    }
                }
                // 写入Hash索引，这样就能通过=，!= 来做查询了
                saveHashIndex(message.getMessage().trim().hashCode(), dataChannel.position());

                byte[] bytes = message.toString().getBytes(charset);
                if (bytes.length < BUFFER_SIZE)
                {
                    buffer.clear();
                    buffer.put(bytes);
                    buffer.flip();
                    dataChannel.write(buffer);
                    buffer.compact();
                } else
                {
                    // 如果数据量过大，就用临时缓冲，否则会发生写入溢出
                    ByteBuffer tmpBuffer = ByteBuffer.allocate(bytes.length + 10);
                    tmpBuffer.put(bytes);
                    dataChannel.write(tmpBuffer);
                    tmpBuffer.compact();
                }
                dataOutputStream.flush();
                wrotenCount += 1;
                printSpeed();
            } catch (IOException e)
            {
                e.printStackTrace();
            }
        }
    }

    // 写入统计
    private int  wrotenCount = 0;
    private long last = System.currentTimeMillis();

    private void printSpeed()
    {
        long now = System.currentTimeMillis();
        long section = now - last;
        if (section > 3000)
        {
            BigDecimal milliSecond = new BigDecimal(section);
            BigDecimal second = milliSecond.divide(new BigDecimal(1000), 5, BigDecimal.ROUND_CEILING);
            BigDecimal times = new BigDecimal(wrotenCount);
            System.out.println("当前写入速度：" + times.divide(second, 5, BigDecimal.ROUND_CEILING).doubleValue() + "/s");
            last = now;
            wrotenCount = 0;
        }
    }

    /**
     * 索引数据缓冲，一条索引的长度固定16个字节，有两个long类型数组组成
     * 分别是时间戳和某条日志在文件存储位置。如果考虑更高性能的话，应该在
     * 压缩一下。
     */
    private ByteBuffer indexBuffer = ByteBuffer.allocate(16); // Long, Long

    private void saveIndex(Long timeStamp, Long position)
    {
        try
        {
            indexBuffer.clear();
            indexBuffer.putLong(timeStamp);
            indexBuffer.putLong(position);
            indexBuffer.flip();
            indexChannel.write(indexBuffer);
            buffer.compact();
            indexOutputStream.flush();
        } catch (IOException e)
        {
            e.printStackTrace();
        }
    }

    private ByteBuffer hashIndexBuffer = ByteBuffer.allocate(12); // int log

    /**
     * 日志内容的Hash索引
     * @param hash 内容的hash值
     * @param position 日志存储的位置
     */
    private void saveHashIndex(int hash, long position)
    {
        try
        {
            hashIndexBuffer.clear();
            hashIndexBuffer.putInt(hash);
            hashIndexBuffer.putLong(position);
            hashIndexBuffer.flip();
            hashIndexChannel.write(hashIndexBuffer);
            hashIndexBuffer.compact();
            hashIndexOutputStream.flush();
        } catch (IOException e)
        {
            e.printStackTrace();
        }
    }

}
